package com.developer.apitest

import org.apache.flink.api.common.functions.{FilterFunction, RichMapFunction}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object TransformTest {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val streamFromFile = env.readTextFile("D:\\idea\\projects\\other\\flink-demo\\src\\main\\resources\\sensor.txt")

    val dataStream: DataStream[SensorReading] = streamFromFile.map( data => {
      val dataArray = data.split(",")
      SensorReading( dataArray(0).trim,dataArray(1).trim.toLong,dataArray(2).trim.toDouble)
    })

    .keyBy(0)
//      .sum(2)
      //输出当前传感器最新的温度+10,而时间戳是上一次数据时间+1
        .reduce( (x,y) => SensorReading( x.id,x.timestamp + 1, y.temperature + 10))

    //2.多流转换算子
    val splitStream = dataStream.split(data => {

      if (data.temperature > 30) Seq("high") else Seq("low")

    })

    //splitStream实现
    val high = splitStream.select("high")
    val low = splitStream.select("low")
    val all = splitStream.select("high","low")

//    high.print("high")
//    low.print("low")
//    all.print("all")
//
//    dataStream.print()

    //合并两条流
//    val warning = high.map( data => (data.id,data.temperature))
//    val connectedStream = warning.connect(low)
//
//    val coMapDataStream = connectedStream.map(
//      warningData => (warningData._1,warningData._2,"warning"),
//      lowData => (lowData.id,"healthy")
//    )
//
//    coMapDataStream.print()

    //合并多条流
//    val unionStream = high.union(low)
//    unionStream.print()

    //函数类
    dataStream.filter(new MyFilter()).print()

    env.execute("transform test")
  }
}

class MyFilter() extends FilterFunction[SensorReading]{
  override def filter(value: SensorReading): Boolean = {
    value.id.startsWith("sensor_1")
  }
}

class MyMapper() extends RichMapFunction[SensorReading,String]{
  override def map(in: SensorReading): String = {

    "flink"
  }

  override def open(parameters: Configuration): Unit = super.open(parameters)
}
